package sunyu.kit.activemq;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.pool.PooledConnection;
import org.apache.activemq.pool.PooledConnectionFactory;
import org.slf4j.Logger;
import sunyu.kit.common.ExceptionKit;
import sunyu.kit.common.LogKit;

import javax.jms.JMSException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * ActiveMQ工具类
 *
 * @author 孙宇
 */
public class ActiveMQKit {

    private static final Logger logger = LogKit.getLogger();

    private static final Map<String, PooledConnection> connectionPool = new ConcurrentHashMap<>();//连接池
    private static final Map<String, ActiveMQSender> senderPool = new ConcurrentHashMap<>();//发送者池
    private static final Map<String, ActiveMQReceiver> receiverPool = new ConcurrentHashMap<>();//接受者池

    /**
     * 项目关闭前调用此方法
     */
    public static void destroy() {
        logger.info("回收ActiveMQ连接开始");
        for (Map.Entry<String, PooledConnection> entry : connectionPool.entrySet()) {
            try {
                entry.getValue().close();
            } catch (JMSException e) {
                logger.error(ExceptionKit.getStackTrace(e));
            }
        }
        logger.info("回收ActiveMQ连接结束");
    }

    /**
     * 添加一个连接
     *
     * @param connectionName 连接名称
     * @param brokerUrl      broker地址
     */
    public static void addConnection(String connectionName,
                                     String brokerUrl) {
        logger.info("初始化ActiveMQ连接，链接名称  {}", connectionName);
        logger.info("brokerUrl {}", brokerUrl);
        if (connectionPool.containsKey(connectionName)) {
            logger.error("ActiveMQ链接  {}  已存在!", connectionName);
        } else {
            ActiveMQConnectionFactory activeMQConnectionFactory = new ActiveMQConnectionFactory();
            activeMQConnectionFactory.setUserName(ActiveMQConnection.DEFAULT_USER);
            activeMQConnectionFactory.setPassword(ActiveMQConnection.DEFAULT_PASSWORD);
            activeMQConnectionFactory.setBrokerURL(brokerUrl);
            activeMQConnectionFactory.setDispatchAsync(true);//异步发送消息
            PooledConnectionFactory pooledConnectionFactory = new PooledConnectionFactory(activeMQConnectionFactory);
            pooledConnectionFactory.setMaximumActiveSessionPerConnection(200);
            pooledConnectionFactory.setIdleTimeout(120);
            pooledConnectionFactory.setMaxConnections(5);
            pooledConnectionFactory.setBlockIfSessionPoolIsFull(true);
            try {
                PooledConnection connection = (PooledConnection) pooledConnectionFactory.createConnection();
                connection.start();
                connectionPool.put(connectionName, connection);
            } catch (JMSException e) {
                logger.error(ExceptionKit.getStackTrace(e));
            }
        }
    }

    /**
     * 获得一个连接
     *
     * @param connectionName 连接名称
     * @return
     */
    public static PooledConnection getConnection(String connectionName) {
        return connectionPool.get(connectionName);
    }


    /**
     * 添加一个生产者
     *
     * @param sender
     */
    public static void addSender(ActiveMQSender sender) {
        logger.info("初始化ActiveMQ Sender {}", sender.getName());
        if (senderPool.containsKey(sender.getName())) {
            logger.error("ActiveMQ Sender {} 已存在!", sender.getName());
        } else {
            senderPool.put(sender.getName(), sender);
        }
    }

    /**
     * 获得一个生产者
     *
     * @param name 生产者名称
     * @return
     */
    public static ActiveMQSender getSender(String name) {
        return senderPool.get(name);
    }

    /**
     * 添加一个消费者
     *
     * @param receiver
     */
    public static void addReceiver(ActiveMQReceiver receiver) {
        logger.info("初始化ActiveMQ Receiver {}", receiver.getName());
        if (receiverPool.containsKey(receiver.getName())) {
            logger.error("ActiveMQ Receiver {} 已存在!", receiver.getName());
        } else {
            receiverPool.put(receiver.getName(), receiver);
        }
    }

    /**
     * 获得一个消费者
     *
     * @param name 消费者名称
     * @return
     */
    public static ActiveMQReceiver getReceiver(String name) {
        return receiverPool.get(name);
    }

}
